import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
import java.math.BigInteger;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.List;

class Map extends Mapper<LongWritable, LongWritable, LongWritable, LongWritable> {
	
	long n = 0;

	/** Reducer setup */
	public void setup (Context context) {
		//Get the value of n from the job configuration
		n = context.getConfiguration().getLong("n", 0);
	}

	
    @Override
    protected void map(LongWritable key, LongWritable value, Context context) throws IOException, InterruptedException {

    	LongWritable new_key; // = new LongWritable(...);
    	LongWritable new_value; // =  new LongWritable(...);
    	long n = key.get();
    	long end = n/value.get();
    	new_value = new LongWritable(end);
    	new_key = new LongWritable(2L);
    	
    	for (int i = 0; i<value.get(); i++){
    		context.write(new_key, new_value);
    		new_key = new LongWritable(new_key.get()+end);
    		new_value = new LongWritable(new_value.get() + end);
    	}
    	
    	
		//key   - n
		//value - into how many ranges the work should be divided
	
		//define content of map
    
    	
    }
}

class Reduce extends Reducer<LongWritable, LongWritable, LongWritable, LongWritable> {
	
	
	BigInteger n;
	
	/** Reducer setup */
	public void setup (Context context) {
		//Get the value of n from the job configuration
		n = new BigInteger(context.getConfiguration().get("n"));
	}

	
    @Override
    protected void reduce(LongWritable key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

    	LongWritable new_key; // = new LongWritable(...);
    	
    	//TODO: Define the REDUCE function
    	for(LongWritable value : values){
    		
		//define content of reduce
    		
    		long start = key.get();
    		long end = value.get();
//    		Long start = new Long("5692479568");
//    		Long end = new Long("6830975475");
    		System.out.println("Start: " + start);
    		System.out.println("End: " + end);
    		while (start <= end) {
    			if (n.mod(BigInteger.valueOf(start)).equals(BigInteger.ZERO)){
    				System.out.println(start);
    				new_key = new LongWritable(start);
    				context.write(new LongWritable(n.longValue()), new_key);
    				return;
    			}
    			start++;
    		}
    	}
    }
    
}

class PollardRho {
    private final static BigInteger ZERO = new BigInteger("0");
    private final static BigInteger ONE  = new BigInteger("1");
    private final static BigInteger TWO  = new BigInteger("2");
    private final static SecureRandom random = new SecureRandom();
    
    
    public static BigInteger rho(BigInteger N) {
        BigInteger divisor;
        BigInteger c  = new BigInteger(N.bitLength(), random);
        BigInteger x  = new BigInteger(N.bitLength(), random);
        BigInteger xx = x;

        // check divisibility by 2
        if (N.mod(TWO).compareTo(ZERO) == 0) return TWO;

        do {
            x  =  x.multiply(x).mod(N).add(c).mod(N);
            xx = xx.multiply(xx).mod(N).add(c).mod(N);
            xx = xx.multiply(xx).mod(N).add(c).mod(N);
            divisor = x.subtract(xx).gcd(N);
        } while((divisor.compareTo(ONE)) == 0);

        return divisor;
    }

    public static BigInteger rhofactor(BigInteger N) {
        if (N.compareTo(ONE) == 0) return ZERO;
        if (N.isProbablePrime(20)) { System.out.println(N); return N; }
        BigInteger divisor = rho(N);
        rhofactor(divisor);
        return rhofactor(N.divide(divisor));
    }
}
    
public class MapReduceSkeletonProject {
    
	
	static Configuration conf;
	
	
	public int run(String pathin, String pathout) throws Exception {
		
		
        
		Job job = new Job(conf);
		job.setJarByClass(MapReduceSkeletonProject.class);

		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);
        
		job.setInputFormatClass(SequenceFileInputFormat.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		
		job.setMapOutputKeyClass(LongWritable.class);
		job.setMapOutputValueClass(LongWritable.class);
		
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(LongWritable.class);	
	
		job.setNumReduceTasks(conf.getInt("nr_of_blocks", 1));
		
		FileInputFormat.addInputPath(job, new Path(pathin));
		FileOutputFormat.setOutputPath(job, new Path(pathout));

		boolean success = job.waitForCompletion(true);
		return success ? 0 : -1;
    }
	

    public static void main(String[] args) throws Exception {
    	
    	MapReduceSkeletonProject skel = new MapReduceSkeletonProject();
    	
    	conf = new Configuration();
    	
    	if(args.length < 3) {
    		System.out.println("Too few arguments. Arguments should be: <hdfs input folder> <hdfs output folder> <num of blocks to divide the computation>");
    		System.out.println("For excample: inputFolderFirstname outputFolderFirstname 5");
    		System.exit(0);
    	}
    	

    	BigInteger n = new BigInteger("21222435582991361");
    	String ns = "21222435582991361";
    	int e = 25;
    	String encrypted_message = "2238 1912 1036 1741 1085 1741 1787 3246 2875 2303 953 3344 1741 1085 2284 1312 1821 2615 684 554 1119 596";
    	
    	
//    	BigInteger n = new BigInteger("129617294548881565397");
//    	String ns = "129617294548881565397";
//    	int e = 65537;
//    	String encrypted_message = "53894636224361534498 46103912565561986894 98347132025044286425 584188771989103 60363371970020293453 84250495626925398392 71188985320803113546 116003784043216197703";
     	
    	String pathin = args[0];
    	String pathout = args[1];
    	int ranges = Integer.parseInt(args[2]);
    	
    	conf.set("n", ns);
    	conf.setLong("nr_of_blocks", ranges);
    	
    	writeNumber(n, ranges, pathin);
//        
    	skel.run(pathin, pathout);

    	ArrayList<Long> factors = readNumber(pathout); 
    	
    	// TESTING ZONEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE
    	BigInteger N = new BigInteger("129617294548881565397");
    	
    	//ArrayList<Boolean> sievelist = new ArrayList<Boolean>();
    	
//    	for (BigInteger i = BigInteger.ONE; i.compareTo(N)==-1; i.add(BigInteger.ONE)){
//    		sievelist.add(true);
//    	}
//    	
//    	boolean sievelist[] = new boolean[1677812737];
//    	for (int i = 0; i<sievelist.length; i++){
//    		sievelist[i] = true;
//    	}
    	
    	
//    	BigInteger nr = new BigInteger("129617294548881565397");
//		System.out.println("fermat: " + factorizefermat(nr));
		
        
		
		
		//Basic factorization
		BigInteger p = BigInteger.valueOf(factors.get(factors.size()-1));
        
        //Correct answer
		//BigInteger p = new BigInteger("6620830889");
		
		//BigInteger p = PollardRho.rhofactor(N);
		
		
		//Function calls to calculate RSA 
		
		String[] full_message = encrypted_message.split(" ");
		String answer = "";
		String[] alphabet = {"","a","b","c","d","e","f","g","h","i","j","k","l","m","n","o","p","q","r","s","t","u","v","w","x","y","z","_"};
		
		
		BigInteger q = n.divide(p);
		System.out.println("p: "+p +"\nq: "+q);
		System.exit(0);
		BigInteger f = p.subtract(BigInteger.ONE).multiply(q.subtract(BigInteger.ONE));
		
		List<BigInteger> eeaList = eea(e, f);
		BigInteger d = modInvEuclid(eeaList, e, f);
		for (int i = 0; i<full_message.length; i++){
			BigInteger x = new BigInteger(full_message[i]);
			BigInteger r = x.modPow(d, n);
			String M = String.valueOf(r);
			while (M.length() > 0) {
				if (M.length()%2 != 0){
					M = "0"+M;
				}
				int index = Integer.parseInt(M.substring(0,2));
				M = M.substring(2);
				answer += alphabet[index];
			}
		}
		System.out.println(answer);
        
    }
    
    
    // Method to use this factorization from another Java class directly
    public static ArrayList<Long>  factor(BigInteger n, long ranges, String  pathin, String pathout) throws Exception {
    	
    	MapReduceSkeletonProject skel = new MapReduceSkeletonProject();
    	
    	conf = new Configuration();
        	
    	conf.set("n", n.toString());
    	conf.setLong("nr_of_blocks", ranges);
    	
    	writeNumber(n, ranges, pathin);
        
        skel.run(pathin, pathout);
        
        ArrayList<Long> factors = readNumber(pathout); 
        
        return factors;
    }
    
    
	/**
	 * Method to write a number to a file on HDFS, acts as an input for MapReduce
	 */
    
    public static void writeNumber (BigInteger number, long ranges,  String pathStr)
	throws IOException
	{
		FileSystem fs = FileSystem.get(conf);
		
		Path path = new Path(pathStr);
		SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, path, 
			LongWritable.class, LongWritable.class, 
			SequenceFile.CompressionType.NONE);
		
		LongWritable index = new LongWritable(Math.round(Math.sqrt(number.doubleValue())+1));
		LongWritable num = new LongWritable(ranges);
		writer.append(index, num);
		writer.close();
	}
    
    
	/**
	 * Method to read numbers (factors) from files on a folder in HDFS
	 */
	public static ArrayList<Long> readNumber (String pathStr) throws IOException
	{
		ArrayList<Long> factors = new ArrayList<Long>();
		
		FileSystem fs = FileSystem.get(conf);
		Path path = new Path(pathStr);
		
		if(!fs.isFile(path)){
			FileStatus[] listFiles = fs.listStatus(path);
			for(int i = 0; i < listFiles.length; i++){
				try{
					SequenceFile.Reader reader = new SequenceFile.Reader(fs, listFiles[i].getPath(), conf);
					LongWritable index = new LongWritable();
					LongWritable el = new LongWritable();
					while (reader.next(index, el)) {
						factors.add(el.get());
					}
					reader.close();
				}
				catch(Exception e){

				}
			}
		}
		return factors;
	}
	
	
	/**
	 * Find prime number
	 * @param n an integer value
	 * @return argument's prime or -1 if no prime found
	 */
	public static int getPrime(int n){
		int i = 2;
		while (i < n) {
			if (n%i == 0){
				return i;
			}
			i++;
		}
		return -1;
	}
	/**
	 * Find greatest common dividor
	 * @param a an integer value
	 * @param b an integer value
	 * @return greatest common dividor of arguments
	 */
	public static int gcd(int a, int b){
		if (b==0){
			return a;
		} else {
			return gcd(b, a%b);
		}
	}
	/**
	 * Find coprimes with extended Euclidean algorithm
	 * @param u an integer value
	 * @param f an integer value
	 * @return List<Integer> containing coprimes
	 */
	public static List<BigInteger> eea(int u, BigInteger f){
		BigInteger u1 = BigInteger.ONE;
		BigInteger u2 = BigInteger.ZERO;
		BigInteger u3 = BigInteger.valueOf(u);
		BigInteger v1 = BigInteger.ZERO;
		BigInteger v2 = BigInteger.ONE;
		BigInteger v3 = f;
		while (!v3.equals(BigInteger.ZERO)){
			BigInteger q = u3.divide(v3);
			BigInteger t1 = u1.subtract(q.multiply(v1));
			BigInteger t2 = u2.subtract(q.multiply(v2));
			BigInteger t3 = u3.subtract(q.multiply(v3));
			u1 = v1;
			u2 = v2;
			u3 = v3;
			v1 = t1;
			v2 = t2;
			v3 = t3;
		}
		List<BigInteger> result = new ArrayList<BigInteger>();
		result.add(u1);
		result.add(u2);
		result.add(u3);
		return result;
		
	}
	/**
	 * Computes the modular multiplicative inverse of a modulo m,
	 * using the extended Euclidean algorithm
	 * @param eeaList an List<Integer> containg coprimes
	 * @param a an integer value
	 * @param f an integer value
	 * @return the modular multiplicative inverse of a modulo m
	 */
	public static BigInteger modInvEuclid(List<BigInteger> eeaList, int a,BigInteger f){
		BigInteger x = eeaList.get(0);
		BigInteger gcd = eeaList.get(2);
		if (gcd.equals(BigInteger.ONE)) {
			return x.mod(f);
		} else {
			return BigInteger.ZERO;
		}
		
	} 
	

	static BigInteger factorizefermat(BigInteger n) {
		BigInteger a, bSq;
		long iterations = 1L;

		a = BigInteger.valueOf((long) Math.sqrt(n.doubleValue()));

		while(iterations<10000000) {
			bSq = a.multiply(a).subtract(n);
			if(isPerfectSquare(bSq.longValue()))
				return a.subtract(BigInteger.valueOf((long) Math.sqrt(bSq.doubleValue()))); // bSq is a square, factorization found
			a = a.add(BigInteger.ONE);
			iterations++;
		}

		return n;
	}
	
	public final static boolean isPerfectSquare(long n)
	{
	    if (n < 0)
	        return false;

	    switch((int)(n & 0xF))
	    {
	    case 0: case 1: case 4: case 9:
	        long tst = (long)Math.sqrt(n);
	        return tst*tst == n;

	    default:
	        return false;
	    }
	}

}

